use std::str::FromStr;

use chrono::{Datelike, Days, NaiveDate};
use redis::Commands;

use super::{get_etl_redis_cli, get_my_pool, SourceType};
use sqlx::Row;
static IMP_LOG_HKEY: &'static str = "file_imp_log";
static SYDT_HKEY: &'static str = "sys_date";
//开始导入日志
pub async fn import_begin_log(
    data_date: &NaiveDate,
    file_name: &str,
    table_name: &str,
    source_type: &SourceType,
) -> anyhow::Result<()> {
    let rcli = get_etl_redis_cli();
    let mut rcon = rcli.get_connection()?;
    let hkey = format!(
        "{IMP_LOG_HKEY}:{source_type}:{}",
        data_date.format("%Y%m%d").to_string()
    );
    let is_ex: bool = rcon.hexists(&hkey, file_name)?;
    let st_str = source_type.to_string();
    if !is_ex {
        sqlx::query("insert into T_IMPORT_LOG(DADT,FLNM,TBNM,BGDT,datp) values(?,?,?,?,?)")
            .bind(data_date)
            .bind(file_name)
            .bind(table_name)
            .bind(&chrono::Local::now())
            .bind(&st_str)
            .execute(get_my_pool())
            .await?;
    } else {
        sqlx::query("update T_IMPORT_LOG set BGDT=? where DADT=? and datp=? and FLNM=?")
            .bind(&chrono::Local::now())
            .bind(data_date)
            .bind(&st_str)
            .bind(file_name)
            .execute(get_my_pool())
            .await?;
    }
    let _: () = rcon.hset(hkey, file_name, false)?;
    Ok(())
}

//完成导入日志
pub async fn import_finish_log(
    data_date: &NaiveDate,
    file_name: &str,
    row_count: u64,
    source_type: &SourceType,
) -> anyhow::Result<()> {
    let st_str = source_type.to_string();
    sqlx::query("update T_IMPORT_LOG set FNDT=?,ROCT=? where DADT=? and datp=? and FLNM=?")
        .bind(&chrono::Local::now())
        .bind(row_count)
        .bind(data_date)
        .bind(st_str)
        .bind(file_name)
        .execute(get_my_pool())
        .await?;
    let rcli = get_etl_redis_cli();
    let mut rcon = rcli.get_connection()?;
    let hkey = format!(
        "{IMP_LOG_HKEY}:{source_type}:{}",
        data_date.format("%Y%m%d").to_string()
    );
    let _: () = rcon.hset(hkey, file_name, true)?;
    Ok(())
}
pub async fn import_is_finish(
    data_date: &NaiveDate,
    file_name: &str,
    source_type: &SourceType,
) -> anyhow::Result<bool> {
    let hkey = format!(
        "{IMP_LOG_HKEY}:{source_type}:{}",
        data_date.format("%Y%m%d").to_string()
    );
    let rcli = get_etl_redis_cli();
    let mut rcon = rcli.get_connection()?;
    let is_ex: bool = rcon.exists(&hkey)?;
    if !is_ex {
        load_import_log(source_type, data_date).await?;
    }
    let is_ok: bool = rcon.hget(hkey, file_name)?;
    Ok(is_ok)
}

async fn load_import_log(source_type: &SourceType, data_date: &NaiveDate) -> anyhow::Result<()> {
    let st_str = source_type.to_string();
    let rows = sqlx::query(
        "select FLNM,case when FNDT is null then 0 else 1 end ISFN from T_IMPORT_LOG where DADT=? and datp=?",
    )
    .bind(&data_date)
    .bind(&st_str)
    .fetch_all(get_my_pool())
    .await?;
    let hkey = format!(
        "{IMP_LOG_HKEY}:{source_type}:{}",
        data_date.format("%Y%m%d").to_string()
    );
    let rcli = get_etl_redis_cli();
    let mut rcon = rcli.get_connection()?;
    for row in rows {
        let file_name: String = row.try_get(0)?;
        let is_fn: i64 = row.try_get(1)?;
        let _: () = rcon.hset(&hkey, file_name, is_fn==1)?;
    }
    let _: () = rcon.expire(hkey, 60 * 60 * 24 * 5)?;
    Ok(())
}
//ImportInfo 导入信息
pub struct ImportInfo {
    pub source_type: SourceType,
    pub data_date: NaiveDate,
}

//获取系统信息
pub async fn get_import_info() -> anyhow::Result<Vec<ImportInfo>> {
    let rows = sqlx::query("select datp,sydt from conf_sydt")
        .fetch_all(get_my_pool())
        .await?;
    let mut res: Vec<ImportInfo> = Vec::new();
    for row in rows {
        let datp: String = row.try_get("datp")?;
        let data_date: NaiveDate = row.try_get("sydt")?;
        let source_type: SourceType = SourceType::from_str(&datp)?;
        res.push(ImportInfo {
            source_type,
            data_date,
        });
    }
    return Ok(res);
}
//根据类型获取系统日期
pub async fn get_sys_date(source_type: SourceType) -> anyhow::Result<NaiveDate> {
    let rcli = get_etl_redis_cli();
    let mut rcon = rcli.get_connection()?;
    let is_ex: bool = rcon.exists(SYDT_HKEY)?;
    if !is_ex {
        let res = get_import_info().await?;
        for item in res {
            let _: () = rcon.hset(
                SYDT_HKEY,
                item.source_type.to_string(),
                item.data_date.format("%Y%m%d").to_string(),
            )?;
        }
    }
    let sys_date_str: String = rcon.hget(SYDT_HKEY, source_type.to_string())?;
    let sys_date = NaiveDate::parse_from_str(&sys_date_str, "%Y%m%d")?;
    return Ok(sys_date);
}
pub async fn source_begin_import(
    data_date: &NaiveDate,
    source_type: &SourceType,
) -> anyhow::Result<()> {
    let st_str = source_type.to_string();
    let _row = sqlx::query("update conf_sydt set isps=1 where datp=?")
        .bind(&st_str)
        .execute(get_my_pool())
        .await?;
    let rcli = get_etl_redis_cli();
    let mut rcon = rcli.get_connection()?;
    let hkey = format!(
        "{IMP_LOG_HKEY}:{source_type}:{}",
        data_date.format("%Y%m%d").to_string()
    );
    let is_ex: bool = rcon.exists(&hkey)?;
    if !is_ex {
        load_import_log(&source_type, data_date).await?;
    }
    return Ok(());
}
pub async fn source_finish_import(
    data_date: &NaiveDate,
    source_type: &SourceType,
    is_one_file: bool,
) -> anyhow::Result<()> {
    let st_str = source_type.to_string();
    if is_one_file {
        let _row = sqlx::query("update conf_sydt set isps=0 where datp=?")
            .bind(&st_str)
            .execute(get_my_pool())
            .await?;
    }else{
        let _row = sqlx::query("update conf_sydt set isps=0,SYDT=? where datp=?")
        .bind(data_date)
        .bind(&st_str)
        .execute(get_my_pool())
        .await?;
        let rcli = get_etl_redis_cli();
        let mut rcon = rcli.get_connection()?;
        let _: () = rcon.hset(SYDT_HKEY, st_str, data_date.format("%Y%m%d").to_string())?;
    }
    return Ok(());
}
pub async fn update_tmst() -> anyhow::Result<()> {
    sqlx::query("update conf_updt set tmst=now() where upid=1").execute(get_my_pool()).await?;
    Ok(())
}
pub async fn get_back_date() -> anyhow::Result<NaiveDate> {
    let row = sqlx::query("select bday,bdtp from conf_updt")
        .fetch_one(get_my_pool())
        .await?;
    let bday: i32 = row.try_get(0)?;
    let bdtp: i8 = row.try_get(1)?;
    let ods_date = get_sys_date(SourceType::ODS).await?;
    if bdtp == 2 && bday > 30 {
        return Err(anyhow::anyhow!(
            "获取回溯日期失败,表conf_updt的字段bdtp为2:按月追溯时，BDAY不能大于30"
        ));
    } else if bdtp == 3 && bday > 90 {
        return Err(anyhow::anyhow!(
            "获取回溯日期失败,表conf_updt的字段bdtp为3:按季度追溯时，BDAY不能大于90"
        ));
    } else if bdtp == 3 && bday > 360 {
        return Err(anyhow::anyhow!(
            "获取回溯日期失败,表conf_updt的字段bdtp为4:按年度追溯时，BDAY不能大于360"
        ));
    }
    let bday=bday as u64;
    let bdate = if bdtp == 1 {
        ods_date.checked_sub_days(Days::new(bday))
    } else if bdtp == 2 {
        let bdate = ods_date.checked_sub_days(Days::new(bday)).unwrap();
        NaiveDate::from_ymd_opt(bdate.year(), bdate.month() - 1, 1)
    } else if bdtp == 3 {
        let bdate = ods_date.checked_sub_days(Days::new(bday)).unwrap();
        let m = bdate.month();
        let m2 = if m <= 3 {
            1
        } else if m <= 6 {
            4
        } else if m <= 9 {
            7
        } else {
            10
        };
        NaiveDate::from_ymd_opt(bdate.year(), m2, 1)
    } else if bdtp == 4 {
        let bdate = ods_date.checked_sub_days(Days::new(bday)).unwrap();
        NaiveDate::from_ymd_opt(bdate.year() - 1, bdate.month(), 1)
    } else {
        None
    };
    let bdate = bdate.ok_or(anyhow::anyhow!(
        "获取回溯日期失败,表bdtp的字段bdtp只能是1:天,2:月,3:季度,4:年度；请检查"
    ))?;
    Ok(bdate)
}
